package com.gugee.ins.data.task.blogger.service;

import com.gugee.common.task.dto.TaskParam;
import com.gugee.common.task.model.TaskRecord;
import com.gugee.ins.common.constant.InsBloggerMsgTypeCodeEnum;
import com.gugee.ins.common.dal.service.blogger.InsBloggerMainDao;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import com.gugee.common.task.service.AbstractProducerHandler;
import com.gugee.ins.data.task.config.TaskConfig;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: luolifeng
 * Date: 2020-12-24 18:32
 */
@Component
@Slf4j
public class InsMBloggerUpdateSvc extends AbstractProducerHandler<InsBloggerMessage> {
    @Autowired
    @Qualifier("insBloggerMsgBaseInfoProducer")
    private KafkaTemplate<Long, InsBloggerMessage> insBloggerMsgBaseInfoProducer;
    @Autowired
    public InsBloggerMainDao insBloggerMainDao;

    @Autowired
    public TaskConfig taskConfig;
    @Override
    public boolean isJump(TaskRecord taskRecord ){
        if(taskRecord.getCurrentPage()!=1) {
            return false;
        }
        /**
         * 判断时间周期,决定是否跳过。一天只发一轮，
         */
        String currentTime= LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
        if(currentTime.equals(taskRecord.getFirstPageStartTime().substring(0,10))){
            XxlJobLogger.log("周期内不重发,跳过， start time:{},currentTime:{}", taskRecord.getFirstPageStartTime(),currentTime);
            log.info("周期内不重发,跳过， start time:{} currentTime:{}", taskRecord.getFirstPageStartTime(),currentTime);
            return true;
        }else{
            return false;
        }


    }
    @Override
    public List<InsBloggerMessage> findMessageObject(TaskRecord taskRecord, TaskParam taskParam) {

        long startId=(taskRecord.getCurrentPage()-1)*taskParam.getRecords()+1L;
        long endId=(long)taskRecord.getCurrentPage()*taskParam.getRecords();
        long lastCollectTime= System.currentTimeMillis() / 1000 - taskConfig.getIntervalHours() * 3600;
        List<InsBloggerMessage> rst= insBloggerMainDao.findInsBloggerMessageByPages(startId,endId,taskConfig.getInsBloggerMainFollowerCountMin(),lastCollectTime);
        return rst==null?new ArrayList<>():rst;
    }

    /**
     * 查询当前有最大BloggerMain的id值，用于代表记录数
     * @return
     */
    @Override
    public long objectTotals() {
        return insBloggerMainDao.getMaxId();
    }

    @Override
    public void sendMsg(List<InsBloggerMessage> messages, String topic) {
        messages.forEach(msg->{
            msg.setRetryTimes(taskConfig.getRetryBloggerTimes());
            msg.setRetryLimitTimes(taskConfig.getRetryBloggerTimes());
            msg.setTypeCode(InsBloggerMsgTypeCodeEnum.UID_UPDATE_F_MBLOGGER_TIMING.getCode());
            insBloggerMsgBaseInfoProducer.send(topic,msg);
        });
        XxlJobLogger.log("投递主题={},投递消息数={}", topic,messages.size());
        log.info("投递主题={},投递消息数={}", topic,messages.size());
    }
}
